package main

import (
	"encoding/csv"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"context"
	"strings"

	"github.com/tsuna/gohbase"
	"github.com/tsuna/gohbase/hrpc"
)

// AppRecord 表示应用记录
type AppRecord struct {
	ID               string
	TrackName        string
	SizeBytes        int64
	Currency         string
	Price            float64
	RatingCountTotal int
	RatingCountVer   int
	UserRating       float64
	UserRatingVer    float64
	Version          string
	ContentRating    string
	PrimeGenre       string
	SupportedDevices int
	IpadScreenShots  int
	Languages        int
	VPP_LICENSE      int
}

const (
	hbaseTable    = "app_records"  // HBase表名
	columnFamily  = "info"         // 列族名
	hbaseZKQuorum = "localhost:2181" // HBase ZooKeeper地址
)

func main() {
	if len(os.Args) != 2 {
		log.Fatal("用法: go run main.go <csv文件路径>")
	}

	csvFile := os.Args[1]
	
	// 读取CSV文件
	records, err := readCSVFile(csvFile)
	if err != nil {
		log.Fatalf("读取CSV文件失败: %v", err)
	}

	// 连接到HBase
	client := gohbase.NewClient(hbaseZKQuorum)
	defer client.Close()

	// 创建表（如果不存在）
	if err := createTableIfNotExists(client); err != nil {
		log.Fatalf("创建表失败: %v", err)
	}

	// 上传数据到HBase
	if err := uploadToHBase(client, records); err != nil {
		log.Fatalf("上传数据到HBase失败: %v", err)
	}

	fmt.Printf("成功上传 %d 条记录到HBase表 '%s'\n", len(records), hbaseTable)
}

// readCSVFile 读取CSV文件并解析为AppRecord切片
func readCSVFile(filename string) ([]AppRecord, error) {
	file, err := os.Open(filename)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	reader := csv.NewReader(file)
	
	// 读取表头（跳过）
	_, err = reader.Read()
	if err != nil {
		return nil, err
	}

	var records []AppRecord
	for {
		line, err := reader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}

		// 解析CSV行到AppRecord结构体
		record, err := parseCSVLine(line)
		if err != nil {
			log.Printf("解析行失败: %v, 跳过该行", err)
			continue
		}

		records = append(records, record)
	}

	return records, nil
}

// parseCSVLine 解析CSV行到AppRecord结构体
func parseCSVLine(line []string) (AppRecord, error) {
	var record AppRecord
	var err error

	if len(line) < 16 {
		return record, fmt.Errorf("行数据不完整，期望16列，实际%d列", len(line))
	}

	record.ID = strings.TrimSpace(line[0])
	record.TrackName = strings.TrimSpace(line[1])
	
	// 解析SizeBytes
	if line[2] != "" {
		record.SizeBytes, err = strconv.ParseInt(line[2], 10, 64)
		if err != nil {
			return record, fmt.Errorf("解析SizeBytes失败: %v", err)
		}
	}

	record.Currency = strings.TrimSpace(line[3])
	
	// 解析Price
	if line[4] != "" {
		record.Price, err = strconv.ParseFloat(line[4], 64)
		if err != nil {
			return record, fmt.Errorf("解析Price失败: %v", err)
		}
	}

	// 解析RatingCountTotal
	if line[5] != "" {
		record.RatingCountTotal, err = strconv.Atoi(line[5])
		if err != nil {
			return record, fmt.Errorf("解析RatingCountTotal失败: %v", err)
		}
	}

	// 解析RatingCountVer
	if line[6] != "" {
		record.RatingCountVer, err = strconv.Atoi(line[6])
		if err != nil {
			return record, fmt.Errorf("解析RatingCountVer失败: %v", err)
		}
	}

	// 解析UserRating
	if line[7] != "" {
		record.UserRating, err = strconv.ParseFloat(line[7], 64)
		if err != nil {
			return record, fmt.Errorf("解析UserRating失败: %v", err)
		}
	}

	// 解析UserRatingVer
	if line[8] != "" {
		record.UserRatingVer, err = strconv.ParseFloat(line[8], 64)
		if err != nil {
			return record, fmt.Errorf("解析UserRatingVer失败: %v", err)
		}
	}

	record.Version = strings.TrimSpace(line[9])
	record.ContentRating = strings.TrimSpace(line[10])
	record.PrimeGenre = strings.TrimSpace(line[11])

	// 解析SupportedDevices
	if line[12] != "" {
		record.SupportedDevices, err = strconv.Atoi(line[12])
		if err != nil {
			return record, fmt.Errorf("解析SupportedDevices失败: %v", err)
		}
	}

	// 解析IpadScreenShots
	if line[13] != "" {
		record.IpadScreenShots, err = strconv.Atoi(line[13])
		if err != nil {
			return record, fmt.Errorf("解析IpadScreenShots失败: %v", err)
		}
	}

	// 解析Languages
	if line[14] != "" {
		record.Languages, err = strconv.Atoi(line[14])
		if err != nil {
			return record, fmt.Errorf("解析Languages失败: %v", err)
		}
	}

	// 解析VPP_LICENSE
	if line[15] != "" {
		record.VPP_LICENSE, err = strconv.Atoi(line[15])
		if err != nil {
			return record, fmt.Errorf("解析VPP_LICENSE失败: %v", err)
		}
	}

	return record, nil
}

// createTableIfNotExists 创建HBase表（如果不存在）
func createTableIfNotExists(client gohbase.Client) error {
	// 检查表是否存在的逻辑需要根据实际需求实现
	// 这里简单返回nil，实际使用时可能需要先创建表
	fmt.Printf("确保表 '%s' 存在...\n", hbaseTable)
	// 实际实现中，你可能需要先通过HBase shell或Admin API创建表
	return nil
}

// uploadToHBase 上传记录到HBase
func uploadToHBase(client gohbase.Client, records []AppRecord) error {
	for i, record := range records {
		// 使用ID作为行键
		rowKey := record.ID
		if rowKey == "" {
			rowKey = fmt.Sprintf("row_%d", i)
		}

		// 创建Put请求
		values := map[string]map[string][]byte{
			columnFamily: {
				"track_name":        []byte(record.TrackName),
				"size_bytes":        []byte(strconv.FormatInt(record.SizeBytes, 10)),
				"currency":          []byte(record.Currency),
				"price":             []byte(strconv.FormatFloat(record.Price, 'f', 2, 64)),
				"rating_count_total": []byte(strconv.Itoa(record.RatingCountTotal)),
				"rating_count_ver":  []byte(strconv.Itoa(record.RatingCountVer)),
				"user_rating":       []byte(strconv.FormatFloat(record.UserRating, 'f', 2, 64)),
				"user_rating_ver":   []byte(strconv.FormatFloat(record.UserRatingVer, 'f', 2, 64)),
				"version":           []byte(record.Version),
				"content_rating":    []byte(record.ContentRating),
				"prime_genre":       []byte(record.PrimeGenre),
				"supported_devices": []byte(strconv.Itoa(record.SupportedDevices)),
				"ipad_screenshots":  []byte(strconv.Itoa(record.IpadScreenShots)),
				"languages":         []byte(strconv.Itoa(record.Languages)),
				"vpp_license":       []byte(strconv.Itoa(record.VPP_LICENSE)),
			},
		}

		putRequest, err := hrpc.NewPutStr(context.Background(), hbaseTable, rowKey, values)
		if err != nil {
			return fmt.Errorf("创建Put请求失败: %v", err)
		}

		_, err = client.Put(putRequest)
		if err != nil {
			return fmt.Errorf("插入数据失败: %v", err)
		}

		if (i+1)%100 == 0 {
			fmt.Printf("已处理 %d/%d 条记录...\n", i+1, len(records))
		}
	}

	return nil
}